cli ¶

def analyse_centrality(
    source_path: Path,
) -> None:
    """
    Identify critical nodes (models) that act as hubs or bridges
    - High in-degree nodes are critical for many models.
    - High betweenness nodes act as bridges between modules.

    """
    graph = build_dependency_graph(source_path=source_path)
    in_degrees = dict(graph.in_degree())
    out_degrees = dict(graph.out_degree())
    betweenness = nx.betweenness_centrality(graph)
    pagerank = nx.pagerank(graph)

    print("Top 5 nodes by in-degree (most dependencies) :")
    for node in sorted(in_degrees, key=in_degrees.get, reverse=True)[:5]:
        print(f"{node}: {in_degrees[node]}")

    print("\nTop 5 nodes by out-degree (most ) :")
    for node in sorted(out_degrees, key=out_degrees.get, reverse=True)[:5]:
        print(f"{node}: {out_degrees[node]}")

    print("\nTop 5 nodes by betweenness centrality (bridges) :")
    for node in sorted(betweenness, key=betweenness.get, reverse=True)[:5]:
        print(f"{node}: {betweenness[node]:.4f}")

    print("\nTop 5 nodes by PageRank (influence) :")
    for node in sorted(pagerank, key=pagerank.get, reverse=True)[:5]:
        print(f"{node}: {pagerank[node]:.4f}")

def analyse_dependency_structure(
    source_path: Path,
) -> None:
    """
    A long critical path suggests a deep dependency chain, which may impact
    performance or testing complexity.

    """
    graph = build_dependency_graph(source_path=source_path)
    if nx.is_directed_acyclic_graph(G=graph):
        topological_order = list(nx.topological_sort(G=graph))
        print("Topological order (first 5):", topological_order[:5])

        # Longest path (critical path)
        longest_path = nx.dag_longest_path(G=graph)
        print(f"Critical path length: {len(longest_path)}")
        print(f"Critical path: {longest_path}")

    else:
        print("Graph is not a DAG. Skipping topological analysis.")

def analyse_graph(
    source_path: Path,
) -> None:
    """
    Calculate fundamental metrics to understand the scale and density of
    dependencies.
    - High node/edge counts suggest complexity.
    - High density (many edges relative to nodes) indicates tightly
      interconnected models, which may complicate maintenance.

    """
    graph = build_dependency_graph(source_path=source_path)
    print(f"Number of nodes (models): {graph.number_of_nodes()}")
    print(f"Number of edges (dependencies): {graph.number_of_edges()}")
    print(f"Graph density: {nx.density(G=graph):.4f}")
    print(
        f"Is the graph a DAG (Directed Acyclic Graph)? {nx.is_directed_acyclic_graph(G=graph)}"
    )

def analyse_path_length(
    source_path: Path,
) -> None:
    """
    Interpretation:
        A low average path length suggests a "small-world" structure, where
        models are reachable in few steps.

    """
    graph = build_dependency_graph(source_path=source_path)

    if nx.is_strongly_connected(G=graph):
        avg_path = nx.average_shortest_path_length(G=graph)
        print(f"Average shortest path length: {avg_path:.2f}")

    else:
        print("Graph is not strongly connected. Calculating for largest component...")
        largest_cc = max(nx.strongly_connected_components(G=graph), key=len)
        G_sub = graph.subgraph(largest_cc)
        avg_path = nx.average_shortest_path_length(G=G_sub)
        print(f"Average shortest path in largest SCC: {avg_path:.2f}")

def detect_cycles_and_strongly_connected_components(
    source_path: Path,
) -> None:
    """
    - Cycles (e.g., A → B → A) indicate potential design flaws.
    - Large Strongly Connected Components suggest tightly interdependent
      modules needing refactoring.

    """
    graph = build_dependency_graph(source_path=source_path)
    try:
        cycles = list(nx.simple_cycles(G=graph))
        print(f"Found {len(cycles)} cycles. Examples:")
        for cycle in cycles[:3]:  # Show first 3
            print(" → ".join(cycle))

    except nx.NetworkXNoCycle:
        print("No cycles detected.")

    strongly_connected_components = list(nx.strongly_connected_components(G=graph))

    print("Strongly Connected Components")
    print(f"Number of components : {len(strongly_connected_components)}")
    print(
        "Largest component :",
        (
            max(strongly_connected_components, key=len)
            if strongly_connected_components
            else "N/A"
        ),
    )

def detect_densely_connected_components(
    source_path: Path,
) -> None:
    """
    Detect clusters of densely connected models to assess modularity


    Interpretation:
    - Communities may align with functional domains (e.g., solar position,
      energy yield, temperature).
    - High modularity suggests well-organized, maintainable code.

    """
    graph = build_dependency_graph(source_path=source_path)
    G_undirected = graph.to_undirected()
    communities = nx_comm.greedy_modularity_communities(G=G_undirected)
    print(f"Detected {len(communities)} communities.")
    for i, community in enumerate(communities):
        print(f"Community {i+1} (size {len(community)}): {community}")

@app.callback()
def main(
    ctx: typer.Context,
    verbose: Annotated[bool, typer.Option(help="Verbose")] = False,
    log_file: Annotated[
        str | None, typer.Option("--log-file", "-l", help="Log file")
    ] = LOG_FILE,
    log_level: str = LOG_LEVEL,
    rich_handler: Annotated[
        bool, typer.Option("--rich", "--no-rich", help="Rich handler")
    ] = RICH_HANDLER,
):
    """
    Inspect data model definitions including YAML files, Python dictionaries
    and native PVGIS data models.
    """
    if verbose:
        log_level = "DEBUG"
    setup_factory_logger(level=log_level, file=log_file, rich_handler=rich_handler)

    # Store logging config in context for child commands
    if not ctx.obj:
        ctx.obj = {}
    ctx.obj["logger_configuration"] = {
        "log_level": log_level,
        "log_file": log_file,
        "rich_handler": rich_handler,
    }

def inspect_pvgis_data_model(
    data_model: str = 'all',
):
    """
    """
    pass

def inspect_python_definition(
    python_module: Path,
    definition: str = "Fingerprint",  # The key to look for in the module's dictionary
    attribute: str | None = None,
    verbose: Annotated[bool, typer.Option(help="Verbose")] = False,
    log_file: Annotated[
        str | None, typer.Option("--log-file", "-l", help="Log file")
    ] = LOG_FILE,
    log_level: str = LOG_LEVEL,
    rich_handler: Annotated[
        bool, typer.Option("--rich", "--no-rich", help="Rich handler")
    ] = RICH_HANDLER,
):
    """Inspect a specific dictionary key from a Python module."""

    # Initialize logging
    setup_factory_logger(
        verbose=verbose,
        level=log_level,
        file=log_file,
        rich_handler=rich_handler,
    )

    if not definition:
        print(f"List all top-level data model definition names")

    else:
        try:
            # Load the data model definitions Python dictionary
            module = load_python_module(python_module)
            definitions = module.PVGIS_DATA_MODEL_DEFINITIONS

            # Extract the specific data model
            if definition in definitions:
                output = definitions[definition]

                if attribute and attribute in output:
                    output = definitions[definition][attribute]

                # Print in YAML format for consistent visualization
                print(
                    yaml.dump(
                        data=output,
                        sort_keys=False,
                        default_flow_style=False,
                        indent=2
                    )
                )
            else:
                logger.warning(f"Key '{definition_key}' not found in the definitions dictionary.")
                raise typer.Exit(code=1)

        except AttributeError:
            typer.echo("The module does not have a 'PVGIS_DATA_MODEL_DEFINITIONS' variable.")
            raise typer.Exit(code=1)

        except Exception as e:
            typer.echo(f"Error: {str(e)}")
            raise typer.Exit(code=1)

def inspect_yaml_definition(
    yaml_file: Path,
    verbose: Annotated[bool, typer.Option(help="Verbose")] = False,
    log_file: Annotated[
        str | None, typer.Option("--log-file", "-l", help="Log file")
    ] = LOG_FILE,
    log_level: str = LOG_LEVEL,
    rich_handler: Annotated[
        bool, typer.Option("--rich", "--no-rich", help="Rich handler")
    ] = RICH_HANDLER,
):
    """ """
    # Initialize logging
    setup_factory_logger(
        verbose=verbose,
        level=log_level,
        file=log_file,
        rich_handler=rich_handler,
    )

    print(
            yaml.dump(
                load_yaml_file(yaml_file),
                sort_keys=False,
            )
        )

def load_python_module(module_path: Path) -> Any:
    """Dynamically load a Python module from a file path."""
    module_name = module_path.stem
    spec = util.spec_from_file_location(module_name, module_path)
    if spec is None:
        raise ImportError(f"Could not load module from {module_path}")

    module = util.module_from_spec(spec)
    spec.loader.exec_module(module)

    return module

def visualise_circular_tree(
    source_path: Annotated[
        Path, typer.Option(help="Source directory with YAML data model descriptions")
    ] = Path(
        "output/complex_example"
    ),  # definitions.yaml"),
    node_size: Annotated[int, typer.Option(help="Node size")] = 20,
) -> None:
    """
    Prototype for the command line
    """
    generate_circular_tree(
        source_path=source_path,
        node_size=node_size,
    )

def visualise_graph(
    source_path: Annotated[
        Path, typer.Option(help="Source directory with YAML data model descriptions")
    ] = Path(
        "output/complex_example"
    ),  # definitions.yaml"),
    # yaml_file: Path = Path("output/complex_example/complex.yaml"),
    node_size: Annotated[int, typer.Option(help="Node size")] = 2400,
    parent_node_size: Annotated[int, typer.Option(help="Parent node size")] = 800,
) -> None:
    """
    Prototype for the command line
    """
    generate_graph(
        source_path=source_path,
        # yaml_file=yaml_file,
        node_size=node_size,
        parent_node_size=parent_node_size,
    )

def visualise_gravis_d3(
    ctx: typer.Context,  # Access parent context
    yaml_file: Annotated[
        Path, typer.Option(help="A YAML PVGIS data model description")
    ] = Path(
        "definitions.yaml"
    ),  # definitions.yaml"),
    # yaml_file: Path = Path("definitions.yaml/data_model_template.yaml"),
    output_file: Path = Path("data_model_graph.html"),
    # node_size: Annotated[int, typer.Option(help="Node size")] = 2400,
    # parent_node_size: Annotated[int, typer.Option(help="Parent node size")] = 800,
) -> None:
    """
    Prototype for the command line
    """
    # Get logging config from parent context if available
    log_config = getattr(ctx.obj, 'log_config', {}) if ctx and ctx.obj else {}
    generate_gravis_d3(
        yaml_file=yaml_file,
        # yaml_file=yaml_file,
        output_file=output_file,
        # node_size=node_size,
        # parent_node_size=parent_node_size,
        **log_config
    )

def visualise_hierarchical_graph(
    source_path: Annotated[
        Path, typer.Option(help="Source directory with YAML data model descriptions")
    ] = Path(
        "output/complex_example"
    ),  # definitions.yaml"),
) -> None:
    """
    Prototype for the command line
    """
    generate_hierarchical_graph(source_path)

def solar_irradiance_introduction():
    """A short introduction on solar irradiance"""
    introduction = """
    [underline]Solar irradiance[/underline] is ...
    """
    note = """
    PVGIS can model solar irradiance components or read selectively
    [magenta]global[/magenta] or [magenta]direct[/magenta] irradiance time series from external datasets.
    """
    from rich.panel import Panel

    note_in_a_panel = Panel(
        "[italic]{}[/italic]".format(note),
        title="[bold cyan]Note[/bold cyan]",
        width=78,
    )
    from rich.console import Console

    console = Console()
    # introduction.wrap(console, 30)
    console.print(introduction)
    console.print(note_in_a_panel)
    console.print(A_PRIMER_ON_SOLAR_IRRADIANCE)

def print_kato_spectral_bands():
    """
    """
    print(KATO_BANDS)

@app.command(
    "physical",
    no_args_is_help=True,
    rich_help_panel=rich_help_panel_irradiance_series,
)
def calculate_physical_limits(
    solar_zenith: float,
    air_temperature: float = 300,
    rounding_places: int = 5,
):
    """Calculate physically possible limits."""
    limits = calculate_limits(solar_zenith, air_temperature, PHYSICALLY_POSSIBLE_LIMITS)
    print_limits_table(limits_dictionary=limits, rounding_places=rounding_places)
    return limits

@app.command(
    "rare",
    no_args_is_help=True,
    rich_help_panel=rich_help_panel_irradiance_series,
)
def calculate_rare_limits(
    solar_zenith: float,
    air_temperature: float = 300,
    rounding_places: int = 5,
):
    """Calculate extremely rare limits."""
    limits = calculate_limits(solar_zenith, air_temperature, EXTREMELY_RARE_LIMITS)
    print_limits_table(limits_dictionary=limits, rounding_places=rounding_places)
    return limits

def print_limits_table(
    limits_dictionary,
    rounding_places=ROUNDING_PLACES_DEFAULT,
):
    """Print table of physically possible irradiance limits"""
    limits_dictionary = round_float_values(limits_dictionary, rounding_places)
    table = Table(box=box.SIMPLE_HEAD)
    table.add_column("Component")
    table.add_column("Min", justify="right")
    table.add_column("Max", justify="right")

    for component, limits in limits_dictionary.items():
        table.add_row(component, str(limits["Min"]), str(limits["Max"]))

    Console().print(table)

@app.command(
    "direct",
    no_args_is_help=True,
    help=f"⦟ Solar incidence angle modifier for direct inclined irradiance due to reflectivity by Martin & Ruiz, 2005 {NOT_COMPLETE_CLI}",
    short_help=f"⦟ Solar incidence angle modifier for direct irradiance due to reflectivity {NOT_COMPLETE_CLI}",
    rich_help_panel=rich_help_panel_toolbox,
)
def get_reflectivity_factor_for_direct_irradiance_series(
    solar_incidence_series: Annotated[
        List[float], typer_argument_solar_incidence_series
    ],
    angular_loss_coefficient: float = ANGULAR_LOSS_COEFFICIENT,
    # csv: Annotated[Path, typer_option_csv] = CSV_PATH_DEFAULT,
    # dtype: Annotated[str, typer_option_dtype] = DATA_TYPE_DEFAULT,
    # array_backend: Annotated[str, typer_option_array_backend] = ARRAY_BACKEND_DEFAULT,
    # uniplot: Annotated[bool, typer_option_uniplot] = UNIPLOT_FLAG_DEFAULT,
    # resample_large_series: Annotated[bool, 'Resample large time series?'] = False,
    # terminal_width_fraction: Annotated[float, typer_option_uniplot_terminal_width] = TERMINAL_WIDTH_FRACTION,
    verbose: Annotated[int, typer_option_verbose] = VERBOSE_LEVEL_DEFAULT,
    log: Annotated[int, typer_option_log] = LOG_LEVEL_DEFAULT,
    # fingerprint: Annotated[bool, typer_option_fingerprint] = FINGERPRINT_FLAG_DEFAULT,
    # metadata: Annotated[bool, typer_option_command_metadata] = False,
    # panels: Annotated[bool, typer_option_panels_output] = False,
    # index: Annotated[bool, typer_option_index] = INDEX_IN_TABLE_OUTPUT_FLAG_DEFAULT,
    quiet: Annotated[bool, typer_option_quiet] = QUIET_FLAG_DEFAULT,
):
    """
    Notes
    -----
    This CLI function uses an implementation of the solar incidence angle
    modifier as per Martin & Ruiz (2005). Expected is the angle between the
    sun-solar-surface vector and the vector normal to the reference solar
    surface. We call this the _typical_ incidence angle as opposed to the
    _complementary_ incidence angle defined by Jenčo (1992).

    """
    reflectivity_factor_for_direct_irradiance_series = (
        calculate_reflectivity_factor_for_direct_irradiance_series(
            solar_incidence_series=solar_incidence_series,
            angular_loss_coefficient=angular_loss_coefficient,
            verbose=verbose,
            log=log,
        )
    )
    if not quiet:
        if verbose > 0:
            pass
        else:
            flat_list = (
                reflectivity_factor_for_direct_irradiance_series.flatten().astype(str)
            )
            csv_str = ",".join(flat_list)
            print(csv_str)

@app.command(
    "indirect",
    no_args_is_help=True,
    help=f"⦟ Solar incidence angle modifier for non-direct inclined irradiance due to reflectivity by Martin & Ruiz, 2005 {NOT_COMPLETE_CLI}",
    short_help=f"⦟ Solar incidence angle modifier for non-direct irradiance due to reflectivity {NOT_COMPLETE_CLI}",
    rich_help_panel=rich_help_panel_toolbox,
)
def get_reflectivity_factor_for_nondirect_irradiance(
    indirect_angular_loss_coefficient: float,
    angular_loss_coefficient: float = ANGULAR_LOSS_COEFFICIENT,
    # csv: Annotated[Path, typer_option_csv] = CSV_PATH_DEFAULT,
    # dtype: Annotated[str, typer_option_dtype] = DATA_TYPE_DEFAULT,
    # array_backend: Annotated[str, typer_option_array_backend] = ARRAY_BACKEND_DEFAULT,
    # uniplot: Annotated[bool, typer_option_uniplot] = UNIPLOT_FLAG_DEFAULT,
    # resample_large_series: Annotated[bool, 'Resample large time series?'] = False,
    # terminal_width_fraction: Annotated[float, typer_option_uniplot_terminal_width] = TERMINAL_WIDTH_FRACTION,
    verbose: Annotated[int, typer_option_verbose] = VERBOSE_LEVEL_DEFAULT,
    log: Annotated[int, typer_option_log] = LOG_LEVEL_DEFAULT,
    # fingerprint: Annotated[bool, typer_option_fingerprint] = FINGERPRINT_FLAG_DEFAULT,
    # metadata: Annotated[bool, typer_option_command_metadata] = False,
    # panels: Annotated[bool, typer_option_panels_output] = False,
    # index: Annotated[bool, typer_option_index] = INDEX_IN_TABLE_OUTPUT_FLAG_DEFAULT,
    quiet: Annotated[bool, typer_option_quiet] = QUIET_FLAG_DEFAULT,
):
    """ """
    reflectivity_factor_for_nondirect_irradiance = (
        calculate_reflectivity_factor_for_nondirect_irradiance(
            indirect_angular_loss_coefficient=indirect_angular_loss_coefficient,
            angular_loss_coefficient=angular_loss_coefficient,
            verbose=verbose,
        )
    )
    if not quiet:
        if verbose > 0:
            pass
        else:
            print(reflectivity_factor_for_nondirect_irradiance)

def introduction():
    """
    A short introduction on the Typical Meteorological Year
    """
    introduction = """The [underline]Typical Meteorological Year[/underline]
    (TMY) is a dataset designed to represent the most _typical_ weather
    conditions for each month at a given location, using historical data. This
    dataset is particularly useful for simulations in solar energy and building
    performance."""
    note = """Internally, [bold]timestamps[/bold] are converted to
    [magenta]UTC[/magenta] and [bold]angles[/bold] are measured in
    [magenta]radians[/magenta] !
    """
    from rich.panel import Panel

    note_in_a_panel = Panel(
        "[italic]{}[/italic]".format(note),
        title="[bold cyan]Note[/bold cyan]",
        width=78,
    )
    from rich.console import Console

    console = Console()
    # introduction.wrap(console, 30)
    console.print(introduction)
    console.print(note_in_a_panel)
    console.print(A_PRIMER_ON_TYPICAL_METEOROLOGICAL_YEAR)

@app.callback()
def main(
    ctx: typer.Context,
    verbose: Annotated[int, typer_option_verbose] = VERBOSE_LEVEL_DEFAULT,
    debug: Annotated[bool, typer.Option(
        "--debug",
        help="Enable debug mode")] = False,
):
    """
    Typical Meteorological Year
    """
    # if verbose > 2:
    #     print(f"Executing command: {ctx.invoked_subcommand}")
    if verbose > 0:
        print("Will output verbosely")
        # state["verbose"] = True

    app.debug_mode = debug

def calculate_degree_days(
    location: Annotated[
        Tuple[float, float], typer.Argument(..., help="Latitude, longitude [°]")
    ],
    years: Annotated[
        Tuple[float, float],
        typer.Argument(
            ..., min=2005, max=2020, help="First and last year of calculations"
        ),
    ],
    meteo: Annotated[
        Path,
        typer.Argument(
            ...,
            exists=True,
            file_okay=True,
            dir_okay=False,
            writable=False,
            readable=True,
            resolve_path=True,
            help="Directory containing the meteorological data",
        ),
    ],
    elevation: Annotated[
        Path,
        typer.Argument(
            ...,
            exists=True,
            file_okay=True,
            dir_okay=False,
            writable=False,
            readable=True,
            resolve_path=True,
            help="Directory containing the digital elevation data",
        ),
    ],
    monthly: bool = typer.Option(False, is_flag=True, help="Print monthly averages"),
):
    """Calculate the cooling/heating degree days

    Args:

    Returns:
        None

    Raises:
        AssertionError: If the command exit code or output doesn't match the expected values.

    Example:

    Notes:
    - Refactored from the original C program `degreedays_hourly` as follows:\n
      - New flag, Old flag, Type, Old Variable, Description\n
      - monthly   , r    , Flag         , -                   , Print monthly averages \n
      - meteo     , g    , str          , dailyPrefix         , Directory containing the meteo files\n
      - elevation , f    , str          , elevationFilename   , Directory prefix of the DEM files\n
      - location  , d    , float, float , latitude, longitude , Latitude, longitude [°]\n
      - years     , y    , float, float , yearStart, yearEnd  , First year, last year of calculations
    """
    pass